package com.niit.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class KafkaProducerTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //1.创建链接
        Properties props = new Properties();
        props.put("bootstrap.servers","node1:9092");
        props.put("acks","all");
        //开启幂等性，保证生产数据的一致性和不可重复性
        props.put("enable.idempotence",true);
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

        //2.创建生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(props);

        //3.发送数据
        for (int i = 1;i<=100;i++){

            ProducerRecord<String,String> bd1 = new ProducerRecord<>("BD1",null,i+"");
            Future<RecordMetadata> send = kafkaProducer.send(bd1);
            //发送成功的应答
            send.get();
            System.out.println("第"+i+"条数据生产成功");

        }

        //4.关闭
        kafkaProducer.close();

    }
}
